package net.bwie.flink.std;

import lombok.SneakyThrows;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;

/**
 * Flink 中窗口Window计算，此处为时间窗口（事件时间）
 * @author xuanyu
 * @date 2025/10/17
 */
public class StreamWindowDemo {

	public static void main(String[] args) throws Exception{
		// 1. 执行环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		// 2. 数据源-source
		DataStreamSource<String> stream = env.socketTextStream("node101", 9999);

		// 3. 数据转换-transformation
		/*
			统计每分钟订单销售额，数据格式如下：
 0       1      2       3
用户ID,商品名称,下单时间,下单金额
u1,milk,2025-10-17 12:00:04,5
u2,smoke,2025-10-17 12:00:34,18
u3,water,2025-10-17 12:00:54,3

u4,bread,2025-10-17 12:01:01,3

u3,water,2025-10-17 12:00:59,2

u4,bread,2025-10-17 12:01:05,5
		 */
		// 3-1. 提取字段值
		SingleOutputStreamOperator<Tuple2<String, Double>> stream31 = stream.map(
			new RichMapFunction<String, Tuple2<String, Double>>() {
				@Override
				public Tuple2<String, Double> map(String value) throws Exception {
					// 字符串分割
					String[] split = value.split(",");
					// 获取字段值
					return Tuple2.of(split[2], Double.valueOf(split[3]));
				}
			}
		);


		/*
			Flink 事件时间窗口计算时步骤如下：
				step1. 指定数据中事件时间字段
					类型Long
				step2. 考虑乱序延迟数据处理
					目前不考虑
				step3. 是否分组
					keyBy
				step4. 设置窗口
					分组：window
					不分组：windowAll
				step5. 窗口计算
					sum
					reduce
		 */
		// todo step1 + step2
		SingleOutputStreamOperator<Tuple2<String, Double>> stream32 = stream31.assignTimestampsAndWatermarks(
			WatermarkStrategy
				// todo 第1道处理：设置窗口触发计算时，等待最大乱序时间，比如设置为5秒
				.<Tuple2<String, Double>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
				// 指定数据中事件时间 -- 必须long类型
				.withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Double>>() {
					@SneakyThrows
					@Override
					public long extractTimestamp(Tuple2<String, Double> element, long recordTimestamp) {
						// 获取订单时间 -> 2025-10-17 12:00:34
						String orderTime = element.f0;
						// 转换long类型
						SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
						Date date = format.parse(orderTime);
						// 返回
						return date.getTime();
					}
				})
		);

		// 迟到数据侧流输出标签
		OutputTag<Tuple2<String, Double>> lateTag = new OutputTag<Tuple2<String, Double>>("late-date"){} ;

		// todo step4
		AllWindowedStream<Tuple2<String, Double>, TimeWindow> stream33 = stream32
			.windowAll(TumblingEventTimeWindows.of(Time.minutes(1)))
			// todo 第2道处理：当窗口触发计算后，可以设置窗口数据保存一段时间，等待某些迟到数据
			.allowedLateness(Time.seconds(10))
			// todo 第3道处理：迟到很久数据，使用侧边流进行输出，后续单独处理
			.sideOutputLateData(lateTag);

		// todo step5
		//SingleOutputStreamOperator<Tuple2<String, Double>> stream3 = stream33.sum(1);

		SingleOutputStreamOperator<String> stream3 = stream33.apply(
			new AllWindowFunction<Tuple2<String, Double>, String, TimeWindow>() {
				SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
				@Override
				public void apply(TimeWindow window,
				                  Iterable<Tuple2<String, Double>> values,
				                  Collector<String> out) throws Exception {
					// 获取窗口开始和结束时间
					String start = format.format(window.getStart());
					String end = format.format(window.getEnd());
					// 窗口中数据计算
					double sum = 0.0 ;
					for (Tuple2<String, Double> value : values) {
						sum += value.f1;
					}
					// 输出
					String output = start + " ~ " + end + ": " + sum ;
					out.collect(output);
				}
			}
		);

		// 获取侧流中迟到数据
		SideOutputDataStream<Tuple2<String, Double>> lateStream = stream3.getSideOutput(lateTag);
		lateStream.print("late");

		// 4.接收器sink
		stream3.print();

		// 5. 触发执行
		env.execute("FlinkWindowWatermarkDemo") ;
	}

}
